/************************************************************************
Copyright 2003, 2004, 2005, 2006, 2007, 2008 Grid Technology Research Center,
National Institute of Advanced Industrial Science and Technology.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
************************************************************************/

package jp.aist.gtrc.plus.scheduler;

import java.rmi.Remote;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.RMIClientSocketFactory;
import java.rmi.server.RMIServerSocketFactory;
import java.rmi.server.UnicastRemoteObject;

import org.apache.log4j.Logger;

import jp.aist.gtrc.plus.reserve.*;
import jp.aist.gtrc.plus.scheduler.node.*;
import jp.aist.gtrc.plus.scheduler.status.*;
import jp.aist.gtrc.plus.scheduler.sys.SchedulerException;

abstract class DefaultScheduler implements PluScheduler {

	protected final SchedulerOptions options;
	protected final StatusManager statusMgr;
	protected final NodeManager nodeMgr;
	protected final ReserveManager rsvMgr;
	protected final ScheduleStarter starter;
	protected final ReserveServer reserveServer;

	protected static final Logger logger = Logger.getLogger(DefaultScheduler.class);

	protected DefaultScheduler(SchedulerOptions options) throws Exception {

		try {
			logger.info("Starting Scheduler");

			this.options = options;

			statusMgr = getStatusManager(options);
			if (statusMgr == null) {
				throw new SchedulerException("StatusManager cannot be initialized");
			}
			nodeMgr = new NodeManager();

			starter = getScheduleStarter(options);
			if (starter == null) {
				throw new SchedulerException("SchedulerStarter cannot be started");
			}

			rsvMgr = getReserveManager(options);
			if (rsvMgr == null) {
				throw new SchedulerException("ReservaManager cannot be initialized");
			}

			reserveServer = rsvMgr.getReserveServer();

			logger.info("Starting Scheduler succeeded");
		} catch (Exception e) {
			logger.error("Starting Scheduler failed");
			logger.debug(e, e);
			stopScheduler();
			throw e;
		}

	}

	protected StatusManager getStatusManager(SchedulerOptions options) throws SchedulerException {

		return StatusManager.getInstance();

	}

	protected ScheduleStarter getScheduleStarter(SchedulerOptions options) throws SchedulerException {

		return new ScheduleStarter();

	}

	protected NodeAllocator getRsvNodeAllocator(SchedulerOptions options) throws SchedulerException {

		return new ReserveNodeAllocator(options, nodeMgr, statusMgr);

	}

	protected ReserveManager getReserveManager(SchedulerOptions options) throws SchedulerException {

		NodeAllocator rsvNodeAllocator = getRsvNodeAllocator(options);
		if (rsvNodeAllocator == null) {
			throw new SchedulerException("Invalid ReserveNodeAllocator");
		}

		ReserveRecorder recorder = new ReserveRecorderImpl(
				options.getReserveFilePath());
		return new ReserveManager(nodeMgr, starter,
				recorder, rsvNodeAllocator, options, statusMgr);

	}

	protected void startReserveServer() throws Exception {

		startReserveServer(null, null);

	}

	protected void startReserveServer(RMIClientSocketFactory csf, RMIServerSocketFactory ssf) throws Exception {

		recoverReserves();

		Remote stub = (ReserveServer)UnicastRemoteObject.exportObject(
				reserveServer, 0, csf, ssf);

		/*
		 * NOTE: The registry must be got by getRegistry(), not by createRegistry()
		 * Both return values are same, but using registry from createRegistry()
		 * and UnicastRemoteObject.exportObject() may not work correctly
		 * because of JDK's bug.
		 * See
		 * http://archives.java.sun.com/cgi-bin/wa?A2=ind0509&L=rmi-users&P=R216&I=-3
		 * http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4114579
		 */
		int rmiPort = options.getRMIPort();
		LocateRegistry.createRegistry(rmiPort);
		Registry registry = LocateRegistry.getRegistry(rmiPort);

		registry.bind(ReserveServer.RMI_OBJECT_NAME, stub);

		logger.info("ReserveServer Started");

                System.err.println("Skip scheduling at start");
//		starter.runSchedule(new ScheduleOrder(getMainServer()));

	}

	protected void stopReserveServer() {

		if (reserveServer != null) {
			try {
				UnicastRemoteObject.unexportObject(reserveServer, true);
				int rmiPort = options.getRMIPort();
				Registry registry = LocateRegistry.getRegistry(rmiPort);
				registry.unbind(ReserveServer.RMI_OBJECT_NAME);
			} catch (Exception e) {
				// ignore
				logger.debug("ReserveServer Exit failed", e);
			}
			logger.info("ReserveServer Exited");
		}

	}

	protected abstract MainServer getMainServer() throws Exception;

	private void recoverReserves() throws Exception {

		if (nodeMgr.getNodeNum() == 0) {
			MainServer srv = getMainServer();
			// get current node status.
			// node status is needed for reserve node check.
			try {
				updateStatus(srv);
			} finally {
				srv.disconnect();
			}
		}

		rsvMgr.recoverFromRecorder();

	}

	protected void stopScheduler() {

		stopReserveServer();
		if (rsvMgr != null) {
			rsvMgr.finish();	// close Reserve information storage
		}

	}

	public void run() {

		logger.info("Scheduler Started, waiting order");

		while(true) {
			ScheduleOrder order = null;
			MainServer srv = null;
			try {
				order = starter.waitOrder();
				if (order == null) {
					assert(false);
					continue;
				}
				if (order.toBeQuit()) {
					break;
				}

				srv = order.getMainServer();
				if (srv == null) {
					srv = getMainServer();
					assert(srv != null);
					order.setMainServer(srv);
				}
				runSchedule(order);
			} catch (Exception e) {
				logger.info(e.getMessage());
				logger.debug(e, e);
			} finally {
				if (srv != null) {
					srv.disconnect();
				}
				if (order != null) {
					order.markAsDone();
				}
			}
		}

		stopScheduler();
		logger.info("Scheduler Exited");

	}

	protected void updateStatus(MainServer srv) throws SchedulerException{

		statusMgr.updateStatus(srv);
		nodeMgr.update(statusMgr.getNodes());

	}

	protected abstract void runSchedule(ScheduleOrder order) throws SchedulerException;

}
